<?php


namespace EchoPHP\Rabbitmq\Query;


use EchoPHP\Facades\Log;
use EchoPHP\Facades\RabbitMq;
use EchoPHP\Rabbitmq\RabbitMqManager;

class QueueProducer {

    protected $rabbitmqConnection;

    public function __construct(string $name) {
        $this->rabbitmqConnection = RabbitMq::connect($name);
    }

    /**
     * 发布消息
     * @param array $messages $messages 发送的消息实体 [['message'=>'消息实体','routekey'=>'路由键(可为空)', 'attributes' => ['消息的附加属性,如消息头类型等']]]
     * @param string $queueName
     * @param bool $confirmSelect
     * @param callable|null $ackCallback
     * @param callable|null $nackCallback
     * @return bool
     * @throws \AMQPChannelException
     * @throws \AMQPConnectionException
     * @throws \AMQPExchangeException
     * @throws \AMQPQueueException
     */
    public function produce(array $messages, string $queueName, bool $confirmSelect = false, callable $ackCallback = null, callable $nackCallback = null) {
        $resultInfo = ['resultAck' => [], 'resultNack' => []];
        if (empty($messages)) {
            return $resultInfo;
        }
        $exchange = $this->rabbitmqConnection->getExchange();
        $exchangeName = $exchange->getName();
        $queue = $this->rabbitmqConnection->getQueue($queueName);
        $queueFlag = $queue->getFlags();
        //判断是否固化队列，如果是，消息也需要固化，否则消息也不需要固化
        $durableQueue = ($queueFlag & AMQP_DURABLE) === AMQP_DURABLE ? true : false;
        //判断是否默认交换机
        $defaultExchange = $this->rabbitmqConnection->isDefaultExchange();

        if ($confirmSelect) {
            //发布者消息确认，打开通道确认并设置回调函数
            $this->rabbitmqConnection->channelConfirm();
            //发送确认模式，定义回调函数
            $this->rabbitmqConnection->getChannel()->setConfirmCallback(function (int $delivery_tag, bool $multiple) use ($ackCallback, $messages, $queue) {
                // ack处理
                //echo 'Message acked', PHP_EOL;
                //var_dump(func_get_args());
                //var_dump($messages[$delivery_tag-1]);

                if (is_callable($ackCallback)) {
                    $ackCallback($delivery_tag, $messages[$delivery_tag - 1]);
                }

                // 看注释：
                //D:/Program Files/JetBrains/PhpStorm 2019.2/plugins/php/lib/php.jar!/stubs/amqp/amqp.php
                //public function confirmSelect()
                //should return boolean **【false】** when wait loop should be canceled.
                //切记 一定要返回 false，否则就阻塞，导致：
                //这条语句：$this->rabbitmqConnection->getChannel()->waitForConfirm(1) 等待超时;
                return false;

            }, function (int $delivery_tag, bool $multiple, bool $requeue) use ($nackCallback, $messages, $queue) {
                // nack处理: 重新发送消息该批次消息，或者记录日志
                //echo 'Message nacked', PHP_EOL;
                //var_dump(func_get_args());
                //var_dump($messages[$delivery_tag-1]);
                if (is_callable($nackCallback)) {
                    $nackCallback($delivery_tag, $messages[$delivery_tag - 1]);
                }

                //切记 一定要返回 false，否则就阻塞，导致：
                //这条语句：$this->rabbitmqConnection->getChannel()->waitForConfirm(1) 等待超时;
                return false;
            });
        }

        $i = 1;
        foreach ($messages as $message) {
            if (is_array($message['message'])) {
                $data = json_encode($message['message']);
            } else {
                $data = (string)$message['message'];
            }

            //使用匿名交换机则路由绑定需要直接指定为队列名称
            $routeKey = isset($message['routekey']) && !empty($message['routekey']) ? $message['routekey'] : '';
            $attributes = $message['attributes'] ?? [];
            $attributes = array_intersect_key($attributes, [
                'content_type' => '', 'content_encoding' => '', 'message_id' => '',
                'user_id' => '', 'app_id' => '', 'delivery_mode' => '',
                'priority' => '', 'timestamp' => '', 'expiration' => '',
                'type' => '', 'reply_to' => '', 'headers' => '',
            ]);
            if ($durableQueue) {
                $attributes['delivery_mode'] = AMQP_DURABLE;
            }

            $result = $exchange->publish($data, $defaultExchange ? $queueName : $routeKey, AMQP_NOPARAM, $attributes);
            if ($confirmSelect) {
                //发送确认模式，阻塞进程等待回调，采用单条确认模式，不适合大批量消息发送
                try {
                    $this->rabbitmqConnection->getChannel()->waitForConfirm(1);
                } catch (\Exception $e) {
                    Log::error('发送消息确认失败:[' . $e->getMessage() . '];', $message);
                }
            }

            $keyStr = $result ? 'resultAck' : 'resultNack';
            $resultInfo[$keyStr][] = ['delivery_tag' => $i, 'message' => $message];
            $i++;
        }

        return $resultInfo;
    }
}